package com.haoniu.system.kafka;

import cn.hutool.json.JSONUtil;
import com.haoniu.common.log.service.AsyncLogService;
import com.haoniu.system.api.domain.SysOperLog;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @ClassName: KafkaListener
 * @Author: ShouwTime
 * @Engineering: haoniu
 * @Date: 2022/7/19  15:52
 * @Version: 1.0
 * @Description: TODO
 */
@Component
public class KafkaListener implements AcknowledgingMessageListener<String,String> {
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    private AsyncLogService asyncLogService;

    @Override
    @org.springframework.kafka.annotation.KafkaListener(topics = "myTopic")
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
        String value = consumerRecord.value();
        SysOperLog sysOperLog = JSONUtil.toBean(value, SysOperLog.class);
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(consumerRecord.key(), "1", 5, TimeUnit.HOURS);
        //表单重复提交
        if (!aBoolean){
            System.err.println("表单重复提交");
        }
        //保存得到数据库
        // 保存数据库
        asyncLogService.saveSysLog(sysOperLog);

        //偏移量
        acknowledgment.acknowledge();
    }
}
